package com.qch.kafka.demo.controller;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

/**
 * @Author qch
 * @Date 2021/5/29
 */
@RestController
public class KafkaAsyncProducerController {
    @Autowired
    private KafkaTemplate<Integer,String>template;

    @RequestMapping("send/async/{message}")
    public String send(@PathVariable String message) throws ExecutionException, InterruptedException {
        ListenableFuture<SendResult<Integer, String>> future = template.send("topic-spring-01", 0, 1, message);
       future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
           @Override
           public void onFailure(Throwable throwable) {
               System.out.println("消息发送失败"+throwable.getMessage());
           }

           @Override
           public void onSuccess(SendResult<Integer, String> integerStringSendResult) {
               RecordMetadata recordMetadata = integerStringSendResult.getRecordMetadata();
               System.out.println("消息发送成功"+recordMetadata.topic()+"\t"+recordMetadata.partition()+"\t"+recordMetadata.offset());
           }
       });
    return "success";
    }
}
